package org.ztest.activemq;

import java.util.concurrent.TimeUnit;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class ActiveMQTest
{
    public void test() throws Exception {
        thread(new TestProducer());
        TimeUnit.SECONDS.sleep(1);
        // 同一条消息只能被一个Consumer消费
        thread(new TestConsumer("ONE"));
        thread(new TestConsumer("TWO"));
        thread(new TestConsumer("THREE"));
        thread(new TestConsumer("FOUR"));
        thread(new TestConsumer("FIVE"));
        thread(new TestConsumer("SIX"));
    }

    private void thread(Runnable runnable) {
        Thread brokerThread = new Thread(runnable);
        brokerThread.start();
    }

    private static class TestProducer implements Runnable
    {
        @Override
        public void run() {
            try {
                // 使用failover前缀可以让客户端达到自动重连的目的
                ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                        "failover:auto://192.168.19.134:61616");

                Connection conn = connectionFactory.createConnection();
                conn.start();

                Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);

                Destination destination = session.createQueue("TEST.FOO");

                MessageProducer producer = session.createProducer(destination);
                producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

                while (true) {
                    String text = "Hello world! From:" + Thread.currentThread().getName() + ":" + this.hashCode();
                    TextMessage message = session.createTextMessage(text);

                    System.out.println("Send message:" + message.hashCode() + ":" + Thread.currentThread().getName());
                    producer.send(message);

//                    TimeUnit.SECONDS.sleep(1);
                }

//                session.close();
//                conn.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    private static class TestConsumer implements Runnable, ExceptionListener
    {
        String name;

        public TestConsumer(String name) {
            this.name = name;
        }

        @Override
        public void run() {
            try {
                ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                        "failover:auto://192.168.19.134:61616");

                Connection conn = connectionFactory.createConnection();
                conn.start();

                conn.setExceptionListener(this);

                Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);

                Destination destination = session.createQueue("TEST.FOO");

                MessageConsumer consumer = session.createConsumer(destination);

                // 使用监听器监听消息
                consumer.setMessageListener(new DefaultMessageListener(name));

                // 这种直接接收消息的方式不建议使用
                // Message message = consumer.receive(1000);
                //
                // if (message instanceof TextMessage) {
                // TextMessage textMessage = (TextMessage) message;
                // String text = textMessage.getText();
                // System.out.println("Received:" + text);
                // } else {
                // System.out.println("Received:" + message);
                // }
                //
                // consumer.close();
                // session.close();
                // conn.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        @Override
        public void onException(JMSException exception) {
            System.out.println("JMS Exception occured. Shutting down client.");
        }
    }

    private static class DefaultMessageListener implements MessageListener
    {
        String name;

        public DefaultMessageListener(String name) {
            this.name = name;
        }

        @Override
        public void onMessage(Message message) {
            System.out.println("In " + name + ", Received:" + message);
        }
    }
}
